package com.xiaoxx.rabbitmq.config;

import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import com.google.common.base.Preconditions;
import com.google.common.base.Splitter;
import com.google.common.collect.Maps;
import com.xiaoxx.rabbitmq.bean.MyMessage;
import com.xiaoxx.rabbitmq.constant.BrokerMessageStatus;
import com.xiaoxx.rabbitmq.constant.MessageType;
import com.xiaoxx.rabbitmq.constant.RabbitQueueExchangeConstant;
import com.xiaoxx.rabbitmq.db.service.MessageStoreService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;

/**
 * $RabbitTemplateContainer 池化封装
 * 每一个topic,对应一个RabbitTemplate
 * 1、提高发送的效率
 * 2、可以根据不同的需求制定不同的RabbitTemplate,比如每一个topic,都有自己的routingKey规则
 * 3, 实现 快速消息 确认消息 可靠消息的基础
 *
 * @author quhiu
 */
@Slf4j
@Component
public class RabbitTemplateContainer implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback {
    private final Map<String /*topic*/, RabbitTemplate> rabbitMap = Maps.newConcurrentMap();

    private final Splitter splitter = Splitter.on("#");
    @Resource
    private CachingConnectionFactory connectionFactory;
    @Resource
    private MessageStoreService messageStoreService;


    public RabbitTemplate getTemplate(MyMessage myMessage) {
        Preconditions.checkNotNull(myMessage);
        String topic = myMessage.getTopic();
        String routingKey = myMessage.getRoutingKey();
        RabbitTemplate rabbitTemplate = rabbitMap.get(topic + routingKey + myMessage.getMessageType());
        if (rabbitTemplate != null) {
            return rabbitTemplate;
        }
        log.info("#新增交换机template模板# topic: {} routingKey:{} 不存在, 创建一个", topic, routingKey);
        RabbitTemplate newTemplate = new RabbitTemplate(connectionFactory);
        // 添加序列化反序列化converter对象
        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.registerModule(new JavaTimeModule());
        newTemplate.setMessageConverter(new Jackson2JsonMessageConverter(objectMapper));
        newTemplate.setExchange(topic);
        newTemplate.setRoutingKey(myMessage.getRoutingKey());
        newTemplate.setRetryTemplate(new RetryTemplate());
        RetryTemplate.builder().maxAttempts(2).build();

        // 开启发布者确认（消息是否到交换机）
        connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        // 开启发布者确认回调 由setPublisherConfirmType控制
        newTemplate.setConfirmCallback(this);

        if (!MessageType.RAPID.equals(myMessage.getMessageType())) {
            // 交换机消息推送确认 （发送的消息是否路由推送到队列）
            newTemplate.setMandatory(true);
            // 交换机消息路由回调 由setMandatory控制
            newTemplate.setReturnsCallback(this);
        }
        rabbitMap.putIfAbsent(topic + routingKey + myMessage.getMessageType(), newTemplate);
        return rabbitMap.get(topic + routingKey + myMessage.getMessageType());
    }


    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String msg) {
        // 具体的应答消息
        List<String> strings = splitter.splitToList(correlationData.getId());
        String messageId = strings.get(0);
        String messageType = strings.get(1);
        if (ack) {
            log.info("消息成功投递到交换机 消息ID: {}", messageId);
            // 如果当前消息类型为reliant 更新消息为发送失败
            if (MessageType.RELIANT.endsWith(messageType)) {
                // 回退回调在 confirm 回调之前 (更新时需判断 防止状态覆盖)
                this.messageStoreService.successByIsStatus(messageId, BrokerMessageStatus.SEND_OK);
            }
        } else {
            log.info("消息未投递到交换机 消息:{} 原因:{}", JSONObject.toJSON(correlationData), msg);
            // 如果当前消息类型为reliant 更新消息为发送失败
            if (MessageType.RELIANT.endsWith(messageType)) {
                this.messageStoreService.updateStatus(messageId, BrokerMessageStatus.SEND_FAIL);
            }
        }
    }

    /**
     * 消息路由失败回调
     */
    @Override
    public void returnedMessage(ReturnedMessage returned) {
        //具体的应答消息
        MyMessage myMessage = Objects.requireNonNull(JSONObject.parseObject(new String(returned.getMessage().getBody()), MyMessage.class));
        // 排除调延迟交换机，因为消息在延迟交换机中延迟，并未送达到队列则出发了此函数回调
        if (!Optional.of(returned).map(ReturnedMessage::getExchange).orElse("").startsWith(RabbitQueueExchangeConstant.DELAY_EXCHANGE_PREFIX)) {
            log.error("消息路由失败：消息:{} 交换机:{} 路由 key:{}",
                    new String(returned.getMessage().getBody()), returned.getExchange(), returned.getRoutingKey());
            // 如果当前消息类型为reliant 就需要到数据库查找并进行更新
            if (MessageType.RELIANT.endsWith(myMessage.getMessageType())) {
                this.messageStoreService.updateStatus(myMessage.getMessageId(), BrokerMessageStatus.SEND_FAIL_ROUTING);
            }
        }

    }
}
